<?php


namespace EchoPHP\Rabbitmq\Query;


use EchoPHP\Facades\RabbitMq;

class QueueConsumer {
    protected $rabbitmqConnection;

    public function __construct(string $name) {
        $this->rabbitmqConnection = RabbitMq::connect($name);
    }

    /**
     * 消费者消费
     * @param callable $dealMessage 消息处理逻辑
     *                                  该函数签名为 function(\AMQPEnvelope $message):bool
     * @param string $queueName 要消费的队列名称
     * @param bool $requeueWhenNack 消费确认失败是否重新入列
     * @param bool $autoAck 是否自动确认  使用自动确认，如果队列消费完成后会自动关闭连接
     * @param string $consumeTag 消费者标识
     * @throws \AMQPChannelException
     * @throws \AMQPConnectionException
     * @throws \AMQPEnvelopeException
     * @throws \AMQPExchangeException
     * @throws \AMQPQueueException
     */
    public function consume(callable $dealMessage, string $queueName = '', bool $requeueWhenNack = false, bool $autoAck = false, string $consumeTag = '') {
        if (!is_callable($dealMessage)) {
            throw new \Exception('消费者回调未定义');
        }
        $queue = $this->rabbitmqConnection->getQueue($queueName);

        $flag = AMQP_NOPARAM;
        if ($autoAck) {
            $flag += AMQP_AUTOACK;
        }

        $callback_func = function (\AMQPEnvelope $message, \AMQPQueue $q) use ($dealMessage, $requeueWhenNack) {
            $dealResult = $dealMessage($message);
            $i = 0;
            $resultFlag = false;
            if (!$dealResult) {
                $resultFlag = false;
                while (!$resultFlag && $i < 2) {
                    $i++;
                    $resultFlag = $q->nack($message->getDeliveryTag(), $requeueWhenNack ? AMQP_REQUEUE : AMQP_NOPARAM);
                }
                return false;
            }

            while (!$resultFlag && $i < 2) {
                $i++;
                $resultFlag = $q->ack($message->getDeliveryTag());
            }

            return true;
        };
        $queue->consume($callback_func, $flag, $consumeTag);
    }

    /**
     * 消费者消费
     * @param callable $dealMessage 消息处理逻辑
     *                                  该函数签名为 function(\AMQPEnvelope $message):bool
     * @param string $queueName 要消费的队列名称
     * @param bool $requeueWhenNack 消费确认失败是否重新入列
     * @param bool $autoAck 是否自动确认  使用自动确认，如果队列消费完成后会自动关闭连接
     * @throws \AMQPChannelException
     * @throws \AMQPConnectionException
     * @throws \AMQPEnvelopeException
     * @throws \AMQPExchangeException
     * @throws \AMQPQueueException
     */
    public function get(callable $dealMessage, string $queueName = '', bool $requeueWhenNack = false, bool $autoAck = false) {
        if (!is_callable($dealMessage)) {
            throw new \Exception('消费者回调未定义');
        }
        $queue = $this->rabbitmqConnection->getQueue($queueName);

        $flag = AMQP_NOPARAM;
        if ($autoAck) {
            $flag += AMQP_AUTOACK;
        }

        $message = $queue->get($flag);

        if (!$message) {
            throw new \Exception('获取' . $queueName . '队列数据失败');
        }

        $deliveryTag = $message->getDeliveryTag();

        $dealResult = $dealMessage($message);
        $i = 0;
        $resultFlag = false;
        if (!$dealResult) {
            while (!$resultFlag && $i < 2) {
                $i++;
                $resultFlag = $queue->nack($message->getDeliveryTag(), $requeueWhenNack ? AMQP_REQUEUE : AMQP_NOPARAM);
            }

            return false;
        }

        //回执结果
        if (!$autoAck) {
            while (!$resultFlag && $i < 2) {
                $i++;
                $resultFlag = $queue->ack($deliveryTag);
            }
        }
        return true;
    }
}